package git.soulbgm.common.serializer;

import git.soulbgm.utils.ProtostuffUtil;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serializer;

import java.util.Map;

/**
 * protostuff serde
 *
 * @author SoulBGM
 * @date 2022/01/16
 */
public class ProtostuffSerde<T> implements Serializer<T>, Deserializer<T>, Serde<T> {

    @Override
    public void configure(final Map<String, ?> configs, final boolean isKey) {
    }

    @SuppressWarnings("unchecked")
    @Override
    public T deserialize(final String topic, final byte[] data) {
        if (data == null) {
            return null;
        }

        UniversalPojo<Object> pojo = ProtostuffUtil.deserializer(data, UniversalPojo.class);
        return (T) pojo.getT();
    }

    @Override
    public byte[] serialize(final String topic, final T data) {
        if (data == null) {
            return null;
        }

        return ProtostuffUtil.serializer(new UniversalPojo<>(data));
    }

    @Override
    public void close() {
    }

    @Override
    public Serializer<T> serializer() {
        return this;
    }

    @Override
    public Deserializer<T> deserializer() {
        return this;
    }

    /**
     * 通用实体
     *
     * @author SoulBGM
     * @date 2024/02/29
     */
    public static class UniversalPojo<T> {
        private T t;

        public UniversalPojo() {
        }

        public UniversalPojo(T t) {
            this.t = t;
        }

        public T getT() {
            return t;
        }
    }

}
